package com.ims.client;

import cn.hutool.core.lang.Console;
import com.ims.client.constant.PulsarUrl;
import com.ims.client.util.*;
import org.apache.pulsar.client.api.*;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

public class ImsClientApplication {

    public static void main(String[] args) {

        String mqIp = "223.83.2.219";
        String mqPort = "6650";

        if (args.length == 2) {
            if (!Pattern.matches("^([1-9]|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])(\\.(\\d|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])){3}$", args[0])) {
                Console.log("输入IP地址格式错误！");
                return;
            }else{
                mqIp = args[0];
                mqPort = args[1];
            }
        } else if (args.length == 1) {
            if (!Pattern.matches("^([1-9]|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])(\\.(\\d|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])){3}$", args[0])) {
                Console.log("输入IP地址格式错误！");
                return;
            }else{
                mqIp = args[0];
            }
        }
        Console.log("访问消息队列地址为 {}:{}", mqIp, mqPort);

        String name = "";
        if ("LINUX".equals(System.getProperty("os.name").toUpperCase())) {
            name = GetDeviceIdUtil.getLinuxDeviceId();
        } else {
            name = GetDeviceIdUtil.getWindowsDeviceId();
        }

        try {
            PulsarClient client = GetPulsarClientUtil.get(mqIp, mqPort);
            Producer<byte[]> producer = client.newProducer()
                    .topic(PulsarUrl.DEVICEINFO_SAVE_TOPIC)
                    .producerName(name)
                    .compressionType(CompressionType.SNAPPY)
                    .sendTimeout(0, TimeUnit.SECONDS)
                    .accessMode(ProducerAccessMode.Shared)
                    .create();

            String msg = EncryptUtil.sm2Encrypt(GenerateDeviceInfoUtil.getAllDeviceInfo().toString());
            CompletableFuture<MessageId> messageIdFuture = producer.sendAsync(msg.getBytes());
            messageIdFuture.whenComplete(((messageId, throwable) -> {
                if (null != throwable) {
                    Console.log("delivery failed, value: " + msg);
                    //此处可以添加延时重试的逻辑
                } else {
                    Console.log("delivered msg " + messageId + ", value:" + msg);
                }
            }));
            Thread.sleep(1000);

            client.close();

        } catch (PulsarClientException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }


        try {
            PulsarClient client = GetPulsarClientUtil.get(mqIp, mqPort);
            Producer<byte[]> simpleProducer = client.newProducer()
                    .topic(PulsarUrl.STATUSINFO_SAVE_TOPIC)
                    .producerName(name)
                    .compressionType(CompressionType.SNAPPY)
                    .sendTimeout(0, TimeUnit.SECONDS)
                    .accessMode(ProducerAccessMode.Shared)
                    .create();

            while (true) {
                String msg = EncryptUtil.sm2Encrypt(GenerateStatusInfoUtil.getAllStatesInfo().toString());
                CompletableFuture<MessageId> messageIdFuture = simpleProducer.sendAsync(msg.getBytes());
                messageIdFuture.whenComplete(((messageId, throwable) -> {
                    if (null != throwable) {
                        Console.log("delivery failed, value: " + msg);
                        //此处可以添加延时重试的逻辑
                    } else {
                        Console.log("delivered msg " + messageId + ", value:" + msg);
                    }
                }));
                Thread.sleep(1000 * 10);
            }
        } catch (PulsarClientException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}